package com.loong.storm.students;

import clojure.main;
import com.loong.commen.util.LogUtil;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

/**
 * storm:数据累加
 * Created by 梁浩峰 on 2016/8/13 12:19.
 */
public class LocalStormTopology implements Serializable{
    public static class DataSourceSpout extends BaseRichSpout  implements Serializable{
        private Map map;
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        /**
         * 本实例运行时执行一次，初始化信息
         * @param map 配置参数
         * @param topologyContext
         * @param spoutOutputCollector 发射器，往下一步发射数据
         */
        public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.map=map;
            this.topologyContext=topologyContext;
            this.spoutOutputCollector=spoutOutputCollector;
        }

        /**
         * 死循环调用
         */
        int i=0;
        public void nextTuple() {
            List<Integer> emit = this.spoutOutputCollector.emit(new Values(i++));
        }

        /**
         * 声明输出内容
         * @param outputFieldsDeclarer
         */
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("number"));
        }
    }


    protected static class Sumbolt extends BaseRichBolt implements Serializable {
        private Map map;
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;

        public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.map=map;
            this.topologyContext=topologyContext;
            this.outputCollector=outputCollector;
        }



        int sum=0;
        /**
         * 死循环读取数据
         * @param tuple
         */
        public void execute(Tuple tuple) {
            Integer number = tuple.getIntegerByField("number");
            sum+=number;
            if(number<50)
            LogUtil.log("spout : "+number +" 加和 :"+sum);
        }

        /**
         * 声明字段
         * @param outputFieldsDeclarer
         */
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

        }
    }

    public  void main(String[] args){
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout("spout_id",new DataSourceSpout());
        topologyBuilder.setBolt("bold_id",new Sumbolt()).shuffleGrouping("spout_id");
        //本地模式
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("topology",new Config(),topologyBuilder.createTopology());

    }

}
